package unit

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/processing/data"
	"github.com/louloulin/dataflare/pkg/processing/processor"
	"github.com/louloulin/dataflare/pkg/processing/storage"
)

// ProcessingUnitConfig 表示处理单元配置
type ProcessingUnitConfig struct {
	// ID 是处理单元ID
	ID string
	// Name 是处理单元名称
	Name string
	// StoragePath 是存储路径
	StoragePath string
	// StorageOptions 是存储选项
	StorageOptions *storage.StorageOptions
	// ProcessorOptions 是处理器选项
	ProcessorOptions *processor.ProcessorOptions
	// BatchSize 是批处理大小
	BatchSize int
	// BatchTimeout 是批处理超时时间
	BatchTimeout time.Duration
}

// DefaultProcessingUnitConfig 返回默认处理单元配置
func DefaultProcessingUnitConfig() *ProcessingUnitConfig {
	return &ProcessingUnitConfig{
		ID:               "default",
		Name:             "Default Processing Unit",
		StoragePath:      "./data",
		StorageOptions:   storage.DefaultStorageOptions(),
		ProcessorOptions: processor.DefaultProcessorOptions(),
		BatchSize:        100,
		BatchTimeout:     time.Second,
	}
}

// ProcessingUnit 表示处理单元
type ProcessingUnit struct {
	// config 是处理单元配置
	config *ProcessingUnitConfig
	// storage 是存储引擎
	storage storage.StorageEngine
	// inputTransport 是输入传输
	inputTransport *data.InMemoryTransport
	// outputTransport 是输出传输
	outputTransport *data.InMemoryTransport
	// processorChain 是处理器链
	processorChain *processor.ProcessorChain
	// pipeline 是数据管道
	pipeline *data.DataPipeline
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
	// wg 是等待组
	wg sync.WaitGroup
}

// NewProcessingUnit 创建一个新的处理单元
func NewProcessingUnit(config *ProcessingUnitConfig) (*ProcessingUnit, error) {
	if config == nil {
		config = DefaultProcessingUnitConfig()
	}

	ctx, cancel := context.WithCancel(context.Background())

	// 创建存储引擎
	store := storage.NewFileStorage(config.StoragePath, config.StorageOptions)

	// 创建传输
	inputTransport := data.NewInMemoryTransport(1000)
	outputTransport := data.NewInMemoryTransport(1000)

	return &ProcessingUnit{
		config:          config,
		storage:         store,
		inputTransport:  inputTransport,
		outputTransport: outputTransport,
		ctx:             ctx,
		cancel:          cancel,
	}, nil
}

// Start 启动处理单元
func (u *ProcessingUnit) Start() error {
	// 打开存储
	err := u.storage.Open(u.ctx)
	if err != nil {
		return fmt.Errorf("failed to open storage: %w", err)
	}

	// 启动传输
	err = u.inputTransport.Start(u.ctx)
	if err != nil {
		u.storage.Close(context.Background())
		return fmt.Errorf("failed to start input transport: %w", err)
	}

	err = u.outputTransport.Start(u.ctx)
	if err != nil {
		u.inputTransport.Stop(context.Background())
		u.storage.Close(context.Background())
		return fmt.Errorf("failed to start output transport: %w", err)
	}

	// 如果处理器链已配置，启动处理器链
	if u.processorChain != nil {
		err = u.processorChain.Start()
		if err != nil {
			u.inputTransport.Stop(context.Background())
			u.outputTransport.Stop(context.Background())
			u.storage.Close(context.Background())
			return fmt.Errorf("failed to start processor chain: %w", err)
		}

		// 连接输入传输到处理器链
		go func() {
			for {
				select {
				case <-u.ctx.Done():
					return
				default:
					// 接收数据
					record, err := u.inputTransport.Receive(u.ctx)
					if err != nil {
						// 检查是否是上下文取消错误
						if u.ctx.Err() != nil {
							return
						}
						// 记录错误并继续
						fmt.Printf("Failed to receive record: %v\n", err)
						time.Sleep(time.Millisecond * 100)
						continue
					}

					// 发送到处理器链
					select {
					case u.processorChain.Input() <- record:
					case <-u.ctx.Done():
						return
					}
				}
			}
		}()

		// 连接处理器链到输出传输
		go func() {
			for {
				select {
				case <-u.ctx.Done():
					return
				case record := <-u.processorChain.Output():
					// 发送到输出传输
					err := u.outputTransport.Send(u.ctx, record)
					if err != nil {
						// 记录错误
						fmt.Printf("Failed to send record to output: %v\n", err)
					}
				}
			}
		}()
	} else {
		// 创建数据管道
		u.pipeline = data.NewDataPipeline(u.inputTransport, u.outputTransport, nil, u.config.BatchSize, u.config.BatchTimeout)

		// 启动数据管道
		err = u.pipeline.Start()
		if err != nil {
			u.inputTransport.Stop(context.Background())
			u.outputTransport.Stop(context.Background())
			u.storage.Close(context.Background())
			return fmt.Errorf("failed to start data pipeline: %w", err)
		}
	}

	// 启动存储写入协程
	u.wg.Add(1)
	go u.storageWriteLoop()

	return nil
}

// Stop 停止处理单元
func (u *ProcessingUnit) Stop() error {
	// 取消上下文
	u.cancel()

	// 等待存储写入协程结束
	u.wg.Wait()

	// 停止数据管道
	if u.pipeline != nil {
		err := u.pipeline.Stop()
		if err != nil {
			return fmt.Errorf("failed to stop data pipeline: %w", err)
		}
	}

	// 停止处理器链
	if u.processorChain != nil {
		err := u.processorChain.Stop()
		if err != nil {
			return fmt.Errorf("failed to stop processor chain: %w", err)
		}
	}

	// 停止传输
	err := u.inputTransport.Stop(context.Background())
	if err != nil {
		return fmt.Errorf("failed to stop input transport: %w", err)
	}

	err = u.outputTransport.Stop(context.Background())
	if err != nil {
		return fmt.Errorf("failed to stop output transport: %w", err)
	}

	// 关闭存储
	err = u.storage.Close(context.Background())
	if err != nil {
		return fmt.Errorf("failed to close storage: %w", err)
	}

	return nil
}

// SetProcessorChain 设置处理器链
func (u *ProcessingUnit) SetProcessorChain(chain *processor.ProcessorChain) {
	u.processorChain = chain
}

// Send 发送数据
func (u *ProcessingUnit) Send(ctx context.Context, record *data.DataRecord) error {
	return u.inputTransport.Send(ctx, record)
}

// SendBatch 发送数据批次
func (u *ProcessingUnit) SendBatch(ctx context.Context, batch *data.DataBatch) error {
	return u.inputTransport.SendBatch(ctx, batch)
}

// Receive 接收数据
func (u *ProcessingUnit) Receive(ctx context.Context) (*data.DataRecord, error) {
	return u.outputTransport.Receive(ctx)
}

// ReceiveBatch 接收数据批次
func (u *ProcessingUnit) ReceiveBatch(ctx context.Context, maxBatchSize int, maxWaitTime time.Duration) (*data.DataBatch, error) {
	return u.outputTransport.ReceiveBatch(ctx, maxBatchSize, maxWaitTime)
}

// Store 存储数据
func (u *ProcessingUnit) Store(ctx context.Context, record *data.DataRecord) error {
	return u.storage.Write(ctx, record)
}

// StoreBatch 存储数据批次
func (u *ProcessingUnit) StoreBatch(ctx context.Context, batch *data.DataBatch) error {
	return u.storage.WriteBatch(ctx, batch)
}

// Retrieve 检索数据
func (u *ProcessingUnit) Retrieve(ctx context.Context, key []byte) (*data.DataRecord, error) {
	return u.storage.Read(ctx, key)
}

// Scan 扫描数据
func (u *ProcessingUnit) Scan(ctx context.Context, startKey, endKey []byte, limit int) ([]*data.DataRecord, error) {
	return u.storage.Scan(ctx, startKey, endKey, limit)
}

// Delete 删除数据
func (u *ProcessingUnit) Delete(ctx context.Context, key []byte) error {
	return u.storage.Delete(ctx, key)
}

// Flush 刷新存储
func (u *ProcessingUnit) Flush(ctx context.Context) error {
	return u.storage.Flush(ctx)
}

// Compact 压缩存储
func (u *ProcessingUnit) Compact(ctx context.Context) error {
	return u.storage.Compact(ctx)
}

// storageWriteLoop 存储写入循环
func (u *ProcessingUnit) storageWriteLoop() {
	defer u.wg.Done()

	for {
		select {
		case <-u.ctx.Done():
			return
		default:
			// 接收数据
			record, err := u.outputTransport.Receive(u.ctx)
			if err != nil {
				// 检查是否是上下文取消错误
				if u.ctx.Err() != nil {
					return
				}
				// 记录错误并继续
				fmt.Printf("Failed to receive record: %v\n", err)
				continue
			}

			// 存储数据
			err = u.storage.Write(u.ctx, record)
			if err != nil {
				// 记录错误
				fmt.Printf("Failed to write record: %v\n", err)
			}
		}
	}
}

// ProcessingUnitManager 表示处理单元管理器
type ProcessingUnitManager struct {
	// units 是处理单元映射表
	units map[string]*ProcessingUnit
	// mu 是互斥锁
	mu sync.RWMutex
}

// NewProcessingUnitManager 创建一个新的处理单元管理器
func NewProcessingUnitManager() *ProcessingUnitManager {
	return &ProcessingUnitManager{
		units: make(map[string]*ProcessingUnit),
	}
}

// CreateUnit 创建处理单元
func (m *ProcessingUnitManager) CreateUnit(config *ProcessingUnitConfig) (*ProcessingUnit, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 检查ID是否已存在
	if _, exists := m.units[config.ID]; exists {
		return nil, fmt.Errorf("processing unit with ID %s already exists", config.ID)
	}

	// 创建处理单元
	unit, err := NewProcessingUnit(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create processing unit: %w", err)
	}

	// 添加到映射表
	m.units[config.ID] = unit

	return unit, nil
}

// GetUnit 获取处理单元
func (m *ProcessingUnitManager) GetUnit(id string) (*ProcessingUnit, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	unit, exists := m.units[id]
	return unit, exists
}

// DeleteUnit 删除处理单元
func (m *ProcessingUnitManager) DeleteUnit(id string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 检查ID是否存在
	unit, exists := m.units[id]
	if !exists {
		return fmt.Errorf("processing unit with ID %s not found", id)
	}

	// 停止处理单元
	err := unit.Stop()
	if err != nil {
		return fmt.Errorf("failed to stop processing unit: %w", err)
	}

	// 从映射表中删除
	delete(m.units, id)

	return nil
}

// ListUnits 列出处理单元
func (m *ProcessingUnitManager) ListUnits() []string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	ids := make([]string, 0, len(m.units))
	for id := range m.units {
		ids = append(ids, id)
	}

	return ids
}

// StartUnit 启动处理单元
func (m *ProcessingUnitManager) StartUnit(id string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 检查ID是否存在
	unit, exists := m.units[id]
	if !exists {
		return fmt.Errorf("processing unit with ID %s not found", id)
	}

	// 启动处理单元
	return unit.Start()
}

// StopUnit 停止处理单元
func (m *ProcessingUnitManager) StopUnit(id string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// 检查ID是否存在
	unit, exists := m.units[id]
	if !exists {
		return fmt.Errorf("processing unit with ID %s not found", id)
	}

	// 停止处理单元
	return unit.Stop()
}

// StartAllUnits 启动所有处理单元
func (m *ProcessingUnitManager) StartAllUnits() error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	for id, unit := range m.units {
		err := unit.Start()
		if err != nil {
			return fmt.Errorf("failed to start processing unit %s: %w", id, err)
		}
	}

	return nil
}

// StopAllUnits 停止所有处理单元
func (m *ProcessingUnitManager) StopAllUnits() error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	for id, unit := range m.units {
		err := unit.Stop()
		if err != nil {
			return fmt.Errorf("failed to stop processing unit %s: %w", id, err)
		}
	}

	return nil
}
